package com.xxd.dfs.namenode;


import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.xxd.dfs.namenode.rpc.model.FetchEditLogResponse;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;
import java.util.Map;

/**
 * @author: XiaoDong.Xie
 * @create: 2020-08-14 09:15
 * @description:
 */
public class FSNameSystem {

    private final static Logger logger = LoggerFactory.getLogger(FSNameSystem.class);

    private FSDirectory directory;

    private FSEditLog fsEditLog;

    private DataNodeManager dataNodeManager;

    private EditLogCache editLogCache = EditLogCache.getInstance();

    private volatile boolean fsRunning = true;
    private NameNodeRpcProto proto = null;

    private volatile long checkpointTxid = 0L;

    public FSNameSystem() {
        this.directory = new FSDirectory();
        this.fsEditLog = new FSEditLog(this);
        this.dataNodeManager = new DataNodeManager(this);
    }

    /**
     * 加载fsimage 和edit log 进来
     *
     * @return
     */
    public String loadFromDisk(Configuration conf) throws IOException {
        String dirPath = conf.get("editlog.dir");
        File dir = new File(dirPath);
        if (!dir.exists()) {
            dir.mkdirs();
            return "";
        }

        // 所有文件
        File[] files = dir.listFiles();
        List<File> editLogFile = new ArrayList<>();
        File fsimage = null;
        for (File file : files) {
            String fileName = file.getName();
            if (fileName.startsWith("fsimage")) {
                fsimage = file;
                this.checkpointTxid =
                        Long.parseLong(fileName.split("_")[1].split("\\.")[0]);
            } else {
                editLogFile.add(file);
            }
        }
        loadFSImage(fsimage);
        loadEditLog(editLogFile, checkpointTxid);

        return fsimage == null ? "" : fsimage.getAbsolutePath();
    }

    /**
     * 加载edit log 文件
     *
     * @param editLogFile
     * @param checkpointTxid
     * @throws IOException
     */
    private void loadEditLog(List<File> editLogFile, long checkpointTxid) throws IOException {
        List<File> files = new ArrayList();
        List<String> fileNames = new ArrayList<>();
        for (File file : editLogFile) {
            fileNames.add(file.getName());
            long endTxid = Long.parseLong(file.getName().split("-")[1].split("\\.")[0]);
            if (endTxid > checkpointTxid) {
                files.add(file);
            }
        }
        fsEditLog.setEditLogFiles(fileNames);

        if (files.isEmpty()) {
            logger.info("当前没有可以恢复的edit log文件.......");
            return;
        }

        Collections.sort(files, new Comparator<File>() {
            @Override
            public int compare(File o1, File o2) {
                int endTxid1 = Integer.parseInt(o1.getName().split("-")[1].split("\\.")[0]);
                int endTxid2 = Integer.parseInt(o2.getName().split("-")[1].split("\\.")[0]);
                return endTxid1 - endTxid2;
            }
        });


        for (File file : files) {
            List<String> editLogs = Files.readAllLines(Paths.get(file.getAbsolutePath()));
            logger.info("正在读取editlog = {} 并进行恢复", file.getName());
            for (String editLogJson : editLogs) {
                EditLog editLog = JSONObject.parseObject(editLogJson, EditLog.class);
                long txid = editLog.getTxid();
                if (txid > checkpointTxid) {
                    // 回放到内存里去
                    String op = editLog.getOp();
                    String path = editLog.getPath();
                    if (op.equals(EditLogOperation.MKDIR)) {
                        logger.info("正在恢复 edit log = {}", editLog);
                        directory.mkdir(path);
                        fsEditLog.setSyTxid(txid);
                        editLogCache.offer(editLog);
                    } else if (op.equals(EditLogOperation.CREATE)) {
                        // directory.create(path);
                    }
                }
            }
        }

    }

    /**
     * 加载fsimage
     *
     * @param file
     */
    private void loadFSImage(File file) {
        if (file == null || !file.exists()) {
            logger.info("没有 fsimage 文件.....");
            return;
        }
        logger.info("正在恢复 fsimage 文件 filename = {}", file.getName());
        FileInputStream inputStream = null;
        FileChannel channel = null;
        try {
            inputStream = new FileInputStream(file);
            channel = inputStream.getChannel();

            ByteBuffer buffer = ByteBuffer.allocate(1024 * 1024); // 1MB 大概能容纳差不多2万多条edit log了
            int count = channel.read(buffer);

            buffer.flip();
            String fsimageJSON = new String(buffer.array(), 0, count);
            FSDirectory.INode root = JSONObject.parseObject(fsimageJSON, FSDirectory.INode.class);
            directory.setRootDir(root);

            fsEditLog.setSyTxid(checkpointTxid);
        } catch (IOException e) {
            logger.error("error", e);
        } finally {
            try {
                if (inputStream != null) {
                    inputStream.close();
                }
                if (channel != null) {
                    channel.close();
                }
            } catch (Exception e) {
                logger.error("error", e);
            }
        }
    }

    /**
     * 创建一个文件目录
     * 1、这是一个绝对路劲
     * 2、不允许创建重复目录
     *
     * @param path 目录
     */
    public boolean mkdir(String path) throws IOException {
        if (directory.mkdir(path)) {
            fsEditLog.logEdit(path, EditLogOperation.MKDIR);
            return true;
        } else {
            logger.warn("文件存在 无法上传哦..........");
            return false;
        }
    }

    /**
     * DN 发起的注册请求
     *
     * @param ip       ip
     * @param hostname 主机名
     * @param nioPort  socket 端口
     */
    public boolean register(String ip, String hostname, int nioPort) {
        DataNodeInfo dataNodeInfo = new DataNodeInfo(ip, hostname, nioPort);
        return dataNodeManager.register(dataNodeInfo);
    }

    /**
     * datanode 发送的心跳请求
     *
     * @param ip
     * @param hostname
     * @return
     */
    public boolean heartbeat(String ip, String hostname) {
        return dataNodeManager.heartbeat(ip, hostname);
    }


    /**
     * 拉取 edit log
     *
     * @param seq 本次拉取的序号
     */
    public FetchEditLogResponse fetchEditLogs(long seq) {
        long ack = 0L;
        List<EditLog> editLogs = editLogCache.fetchEditLogs();
        if (editLogs.isEmpty()) {
            ack = seq;
        } else {
            ack = seq + editLogs.size();
        }

        String editlogsJSON = JSONArray.toJSONString(editLogs);

        FetchEditLogResponse response = FetchEditLogResponse.newBuilder()
                .setAck(ack)
                .setEditLogs(editlogsJSON)
                .build();
        return response;
    }

    public Map<String, DataNodeInfo> getNodeInfoMap() {
        return dataNodeManager.getDatanodes();
    }


    public void printTree() {
        directory.printlnTree();
    }


    /**
     * 更新checkpoint Txid
     *
     * @param txid
     */
    public void updateCheckpointTxid(long txid) {
        this.checkpointTxid = txid;
    }

    /**
     * 获取Checkpoint Txid
     *
     * @return
     */
    public long getCheckpointTxid() {
        return checkpointTxid;
    }

    public void setProto(NameNodeRpcProto proto) {
        this.proto = proto;
    }

    /**
     * 优雅关闭
     * 1、把edit log 强制刷入
     */
    public void shutdown() {
        logger.info("收到 关闭 namenode 请求, 准备进行优雅关闭........");
        try {
            fsRunning = false;
            fsEditLog.forceFlush();
            proto.stop();

            System.exit(1);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 是否正在运行
     *
     * @return
     */
    public boolean isRunning() {
        return fsRunning;
    }

    /**
     * 创建文件
     *
     * @param filename 文件名称
     * @return
     */
    public boolean create(String filename) throws IOException {
        if (!mkdir(filename)) {
            logger.info("文件上传失败...........");
            return false;
        }
        return true;
    }

    /**
     * 为此文件上传分配datanode
     *
     * @param filename 文件名称
     * @param fileSize 文件大小
     */
    public String allocateDataNodes(String filename, long fileSize) {
        List<DataNodeInfo> dataNodeInfos = dataNodeManager.allocateDataNodes(fileSize);
        String dataNodesJson = JSON.toJSONString(dataNodeInfos);
        logger.info("给{} 分配的 datanode = 【{}】", filename, dataNodesJson);
        return dataNodesJson;
    }
}
